package cn._51doit.day02;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

import java.util.Properties;

/**
 * 标准键盘布局
 * ┌───┐   ┌───┬───┬───┬───┐ ┌───┬───┬───┬───┐ ┌───┬───┬───┬───┐ ┌───┬───┬───┐
 * │Esc│   │ F1│ F2│ F3│ F4│ │ F5│ F6│ F7│ F8│ │ F9│F10│F11│F12│ │P/S│S L│P/B│  ┌┐    ┌┐    ┌┐
 * └───┘   └───┴───┴───┴───┘ └───┴───┴───┴───┘ └───┴───┴───┴───┘ └───┴───┴───┘  └┘    └┘    └┘
 * ┌───┬───┬───┬───┬───┬───┬───┬───┬───┬───┬───┬───┬───┬───────┐ ┌───┬───┬───┐ ┌───┬───┬───┬───┐
 * │~ `│! 1│@ 2│# 3│$ 4│% 5│^ 6│& 7│* 8│( 9│) 0│_ -│+ =│ BacSp │ │Ins│Hom│PUp│ │N L│ / │ * │ - │
 * ├───┴─┬─┴─┬─┴─┬─┴─┬─┴─┬─┴─┬─┴─┬─┴─┬─┴─┬─┴─┬─┴─┬─┴─┬─┴─┬─────┤ ├───┼───┼───┤ ├───┼───┼───┼───┤
 * │ Tab │ Q │ W │ E │ R │ T │ Y │ U │ I │ O │ P │{ [│} ]│ | \ │ │Del│End│PDn│ │ 7 │ 8 │ 9 │   │
 * ├─────┴┬──┴┬──┴┬──┴┬──┴┬──┴┬──┴┬──┴┬──┴┬──┴┬──┴┬──┴┬──┴─────┤ └───┴───┴───┘ ├───┼───┼───┤ + │
 * │ Caps │ A │ S │ D │ F │ G │ H │ J │ K │ L │: ;│" '│ Enter  │               │ 4 │ 5 │ 6 │   │
 * ├──────┴─┬─┴─┬─┴─┬─┴─┬─┴─┬─┴─┬─┴─┬─┴─┬─┴─┬─┴─┬─┴─┬─┴────────┤     ┌───┐     ├───┼───┼───┼───┤
 * │ Shift  │ Z │ X │ C │ V │ B │ N │ M │< ,│> .│? /│  Shift   │     │ ↑ │     │ 1 │ 2 │ 3 │   │
 * ├─────┬──┴─┬─┴──┬┴───┴───┴───┴───┴───┴──┬┴───┼───┴┬────┬────┤ ┌───┼───┼───┐ ├───┴───┼───┤ E││
 * │ Ctrl│    │Alt │         Space         │ Alt│    │    │Ctrl│ │ ← │ ↓ │ → │ │   0   │ . │←─┘│
 * └─────┴────┴────┴───────────────────────┴────┴────┴────┴────┘ └───┴───┴───┘ └───────┴───┴───┘
 *
 * @create: 2021-10-15 14:21
 * @author: 今晚打脑斧
 * @program: KafkaSource
 * @Description:
 *   使用KafkaSource从Kafka消息队列中读取数据
 *     1.KafkaSource创建的DataStream是一个并行的DataStream
 *     2.KafkaSource创建的DataStream是一个无限的数据流
 *
 *   使用步骤：
 *    https://nightlies.apache.org/flink/flink-docs-release-1.13/docs/connectors/datastream/kafka/
 *
 *    1.导入依赖：
 *        <dependency>
 *          <groupId>org.apache.flink</groupId>
 *          <artifactId>flink-connector-kafka_2.11</artifactId>
 *          <version>1.13.2</version>
 *        </dependency>
 *
 *    2.new FlinkKafkaConsumer
 *
 *    3.调用env的addSource传入FlinkKafkaConsumer实例
 *
 *    生产者的数据轮询的放到分区里面(生产者也可以指定key和分区编号)(相同key的会到同一个分区里面,但是同一分区内会有其它的key)
 *    消费者在会从分区里面读取数据,一个消费者组里面,每个消费者都不会读另一个消费者的分区数据
 *    但是同一消费者组里消费者个数少于分区数,那么一个消费者会读取多个分区数据
 **/
public class KafkaSource {
    public static void main(String[] args)throws Exception {
        Configuration configuration = new Configuration();
        //指定本地web页面的服务的端口号
        configuration.setInteger("rest.port", 8081);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(configuration);

        /**
         * topic在创建的时候有3个分区(每个分区都会存储数据,别和副本概念混了),
         * 而程序执行的时候,并行度为4,Source默认的并行度也为4.
         * 在同一个 消费者组里面,不能同时由多个task读取 数据,同时更改偏移量,会出问题.
         * 所以,会有算子task闲下来
         */
        //设置Evn的并行度,证明会有一个task读取两个分区的内容
        env.setParallelism(2);

        Properties properties = new Properties();

        //服务器
        properties.setProperty("bootstrap.servers", "doit01:9092,doit02:9092,doit03:9092");
        //消费者组ID
        properties.setProperty("group.id", "test777");
        //从头读取数据
        properties.setProperty("auto.offset.reset", "earliest"); //如果没有记录历史偏移量就从头读


        FlinkKafkaConsumer<String> flinkKafkaConsumer = new FlinkKafkaConsumer<>(
                "wordcount",           //topic
                new SimpleStringSchema(),   //反序列化方式
                properties                  //配置文件
        );

        //调用env的addSource方法创建DataStream
        DataStreamSource<String> lines = env.addSource(flinkKafkaConsumer); //上面设置了env的并行度为2,所以它的并ing的也为2,
                                                                            // flinkKafkaConsumer并行度默认是与env并行度一致的

        //kafkaSource的并行度为2,代表着有test777这个消费者组里面有四个消费者,但是分区只有三个,所以会有一个消费者不会消费数据
        System.out.println("kafkaSource的并行度：" + lines.getParallelism());

        lines.print();

        env.execute();

    }
}
